package com.atguigu.flink.cep;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/3
 *
 *  多个模式，使用某种关系连接。
 *
 *      模式1  关系  模式2  关系 模式3
 */
public class Demo4_PatternSequence
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, r) -> e.getTs());

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds =
            env//.socketTextStream("hadoop103",8888)
                .readTextFile("data/cep.txt")
               .map(new WaterSensorMapFunction())
               .assignTimestampsAndWatermarks(watermarkStrategy);


        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("规则1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            })
            // 上下两个模式应用的数据必须紧挨着。  当前数据以规则1开头(id=s1),接下来(next)那条数据符合规则2(id=s2),满足，输出当前这条和接下来那条
            .next("规则2")
            // 上下两个模式应用的数据不必紧挨着。  当前数据以规则1开头(id=s1),接下来后续的那条数据符合规则2(id=s2),满足，输出当前这条和接下来那条
            //.followedBy("规则2")
            // 上下两个模式应用的数据必须不能紧挨着。  当前数据以规则1开头(id=s1),接下来(next)那条数据不符合规则2(id=s2),满足，输出当前这条和接下来那条
            //.notNext("规则2")
            // 上下两个模式应用的数据不必紧挨着。  当前数据以规则1开头(id=s1),接下来后续不能有符合规则2(id=s2)的数据,满足，输出当前这条和接下来那条
            //NotFollowedBy is not supported as a last part of a Pattern!  不能作为pattern的最后一部分
            //.notFollowedBy("规则2")
            // 滥交模式。 当前数据必须是s1会和后续到达的任意一个s2 匹配
            //.followedByAny("规则2")
            .where(new SimpleCondition<WaterSensor>()
            {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s2".equals(value.getId());
                }
            })
            /*.followedBy("规则3")
            .where(new SimpleCondition<WaterSensor>()
            {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s3".equals(value.getId());
                }
            })*/;


        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);


        patternStream.select(new PatternSelectFunction<WaterSensor, String>()
        {

            @Override
            public String select(Map<String, List<WaterSensor>> map) throws Exception {
                return map.toString();
            }
        })
                     .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
